package com.mq.mqClient;

import com.mq.common.*;
import com.mq.mqserver.core.BasicProperties;
import com.mq.mqserver.core.ExchangeType;
import lombok.Data;

import java.io.IOException;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 定义一个通道
 */
@Data
public class Channel {
    //通道的唯一标识
    private String channelId;
    //当前channel属于哪个连接
    private Connection connection;
    //使用hash来存储客户端收到的服务器相应
    //key :rid  ,  val:返回的结果
    ConcurrentHashMap<String, BasicReturns> basicReturnsMap = new ConcurrentHashMap<>();
    //当一个channel订阅了某个队列的时候,就要记录对应的回调是什么,
    //当队列返回消息的时候,执行回调
    //此处约定,一个消息只有一个回调
    private Consumer consumer = null;

    public Channel(String channelId,Connection connection){
        this.channelId = channelId;
        this.connection = connection;
    }

    /**   type 表⽰请求响应不同的功能. 取值如下
     *  0x1  创建 channel
     * • 0x2  关闭 channel
     * • 0x3  创建 exchange
     * • 0x4  销毁 exchange
     * • 0x5  创建 queue
     * • 0x6  销毁 queue
     * • 0x7  创建 binding
     * • 0x8  销毁 binding
     * • 0x9  发送 message
     * • 0xa  订阅 message
     * • 0xb  返回 ack
     * • 0xc  服务器给客⼾端推送的消息. (被订阅的消息) 响应独有的
     */
//  和服务器交互,告知服务器,客户端创建了一个新的channel
    public boolean createChannel() throws IOException {
        //构造请求数据
        BasicArgs basicArgs = new BasicArgs();
        basicArgs.setRid(generateRid());
        basicArgs.setChannelId(channelId);
        byte[] payload = BinaryTool.toByte(basicArgs);

        //构造请求对象
        Request request = new Request();
        request.setType(0x1);
        request.setLength(payload.length);
        request.setPayload(payload);

        //将请求发送给服务器
        connection.writeRequest(request);

        //等待服务器的针对该请求的响应,可能会阻塞
        BasicReturns basicReturns = waitResponse(basicArgs.getRid());
        return basicReturns.isOk();
    }

//  阻塞等待服务器返回的响应
    private BasicReturns waitResponse(String rid) {
        //1.获取响应
        BasicReturns basicReturns = null;
        //这里要不停的判断,要用while语句
        while((basicReturns = basicReturnsMap.get(rid)) == null){
            //当未查到时,说明响应还未返回,需要阻塞等待
            synchronized (this){
                try {
                    wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
        //读取成功后,还要将响应从hash表中删除
        basicReturnsMap.remove(rid);
        //2.返回响应
        return basicReturns;
    }

    //将响应存到结果集basicReturnsMap中
    public void putReturnsMap(BasicReturns basicReturns) {
        basicReturnsMap.put(basicReturns.getRid(),basicReturns);
        synchronized (this){
            //唤醒所有处于阻塞状态的线程
            notifyAll();
        }
    }

//     生成rid,请求的唯一标识
    private String generateRid() {
        return "R-"+ UUID.randomUUID();
    }
    //关闭channel:0x2
    public boolean close() throws IOException {
        //构造请求数据
        BasicArgs basicArgs = new BasicArgs();
        basicArgs.setRid(generateRid());
        basicArgs.setChannelId(channelId);
        byte[] payload = BinaryTool.toByte(basicArgs);

        //构造请求对象
        Request request = new Request();
        request.setType(0x2);
        request.setLength(payload.length);
        request.setPayload(payload);

        //将请求发送给服务器
        connection.writeRequest(request);

        //等待服务器的针对该请求的响应,可能会阻塞
        BasicReturns basicReturns = waitResponse(basicArgs.getRid());
        return basicReturns.isOk();
    }

//创建交换机 0x3
    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType,
                               boolean isDurable, boolean autoDelete, Map<String,Object> args) throws IOException {

        ExchangeDeclareArgs exchangeDeclareArgs = new ExchangeDeclareArgs();
        exchangeDeclareArgs.setRid(generateRid());
        exchangeDeclareArgs.setChannelId(channelId);
        exchangeDeclareArgs.setExchangeName(exchangeName);
        exchangeDeclareArgs.setType(exchangeType);
        exchangeDeclareArgs.setDurable(isDurable);
        exchangeDeclareArgs.setAutoDelete(autoDelete);
        exchangeDeclareArgs.setArgs(args);
        byte[] payload = BinaryTool.toByte(exchangeDeclareArgs);

        //构造请求对象
        Request request = new Request();
        request.setType(0x3);
        request.setLength(payload.length);
        request.setPayload(payload);

        //发送请求 给服务器
        connection.writeRequest(request);
        //等待响应
        BasicReturns returns = waitResponse(exchangeDeclareArgs.getRid());
        return returns.isOk();

    }
        //删除交换机 0x4
        public boolean exchangeDelete(String exchangeName) throws IOException {
            ExchangeDeleteArgs exchangeDeleteArgs = new ExchangeDeleteArgs();
            exchangeDeleteArgs.setRid(generateRid());
            exchangeDeleteArgs.setChannelId(channelId);
            exchangeDeleteArgs.setExchangeName(exchangeName);
            byte[] payload = BinaryTool.toByte(exchangeDeleteArgs);
            //构造
            Request request = new Request();
            request.setType(0x4);
            request.setLength(payload.length);
            request.setPayload(payload);

            //发送
            connection.writeRequest(request);
            //等待响应
            BasicReturns returns = waitResponse(exchangeDeleteArgs.getRid());
            return returns.isOk();
        }


        //创建队列   0x5
        public boolean queueDeclare(String queueName,boolean isDurable,boolean autoDelete,
                                    boolean exclusive,Map<String,Object> args) throws IOException {
            QueueDeclareArgs queueDeclareArgs = new QueueDeclareArgs();
            queueDeclareArgs.setRid(generateRid());
            queueDeclareArgs.setChannelId(channelId);
            queueDeclareArgs.setQueueName(queueName);
            queueDeclareArgs.setDurable(isDurable);
            queueDeclareArgs.setAutoDelete(autoDelete);
            queueDeclareArgs.setExclusive(exclusive);
            queueDeclareArgs.setArgs(args);
            byte[] payload = BinaryTool.toByte(queueDeclareArgs);

            Request request = new Request();
            request.setType(0x5);
            request.setLength(payload.length);
            request.setPayload(payload);

            //发送
            connection.writeRequest(request);
            //等待结果
            BasicReturns returns = waitResponse(queueDeclareArgs.getRid());
            return returns.isOk();
        }

        //删除队列: 0x6
        public boolean queueDelete(String queueName) throws IOException {
            QueueDeleteArgs queueDeleteArgs = new QueueDeleteArgs();
            queueDeleteArgs.setRid(generateRid());
            queueDeleteArgs.setChannelId(channelId);
            queueDeleteArgs.setQueueName(queueName);
            byte[] payload = BinaryTool.toByte(queueDeleteArgs);

            Request request = new Request();
            request.setType(0x6);
            request.setLength(payload.length);
            request.setPayload(payload);

            connection.writeRequest(request);
            BasicReturns basicReturns = waitResponse(queueDeleteArgs.getRid());
            return basicReturns.isOk();
        }


        //添加绑定 :0x7
        public boolean bindingDeclare(String exchangeName,String queueName,String bindingKey) throws IOException {
            BindingDeclareArgs bindingDeclareArgs = new BindingDeclareArgs();
            bindingDeclareArgs.setRid(generateRid());
            bindingDeclareArgs.setChannelId(channelId);
            bindingDeclareArgs.setExchangeName(exchangeName);
            bindingDeclareArgs.setQueueName(queueName);
            bindingDeclareArgs.setBindingKey(bindingKey);
            byte[] payload = BinaryTool.toByte(bindingDeclareArgs);

            Request request = new Request();
            request.setType(0x7);
            request.setLength(payload.length);
            request.setPayload(payload);

            connection.writeRequest(request);

            BasicReturns basicReturns = waitResponse(bindingDeclareArgs.getRid());
            return basicReturns.isOk();
        }

        //删除绑定: 0x8
        public boolean bindingDelete(String exchangeName,String queueName) throws IOException {
            BindingDeleteArgs bindingDeleteArgs = new BindingDeleteArgs();
            bindingDeleteArgs.setRid(generateRid());
            bindingDeleteArgs.setChannelId(channelId);
            bindingDeleteArgs.setExchangeName(exchangeName);
            bindingDeleteArgs.setQueueName(queueName);
            byte[] payload = BinaryTool.toByte(bindingDeleteArgs);

            Request request = new Request();
            request.setType(0x8);
            request.setLength(payload.length);
            request.setPayload(payload);

            connection.writeRequest(request);
            BasicReturns basicReturns = waitResponse(bindingDeleteArgs.getRid());
            return basicReturns.isOk();
        }

        //发送消息 0x9
        public boolean basicPublish(String exchangeName, String routingKey, BasicProperties basicProperties, byte[] body) throws IOException {
            BasicPublishArgs basicPublishArgs = new BasicPublishArgs();
            basicPublishArgs.setRid(generateRid());
            basicPublishArgs.setChannelId(channelId);
            basicPublishArgs.setExchangeName(exchangeName);
            basicPublishArgs.setRoutingKey(routingKey);
            basicPublishArgs.setBasicProperties(basicProperties);
            basicPublishArgs.setBody(body);
            byte[] payload = BinaryTool.toByte(basicPublishArgs);

            Request request = new Request();
            request.setType(0x9);
            request.setLength(payload.length);
            request.setPayload(payload);

            connection.writeRequest(request);
            BasicReturns basicReturns = waitResponse(basicPublishArgs.getRid());
            return basicReturns.isOk();
        }


        //订阅消息 :0xa
        public boolean basicConsumer(String queueName, boolean autoAck, Consumer consumer) throws IOException, MqException {
            // 先设置回调.
            if (this.consumer != null) {
                throw new MqException("该 channel 已经设置过消费消息的回调了, 不能重复设置!");
            }
            this.consumer = consumer;

            BasicConsumerArgs basicConsumerArgs = new BasicConsumerArgs();
            basicConsumerArgs.setRid(generateRid());
            basicConsumerArgs.setChannelId(channelId);
            basicConsumerArgs.setQueueName(queueName);
            basicConsumerArgs.setAutoAck(autoAck);
            byte[] payload = BinaryTool.toByte(basicConsumerArgs);

            Request request = new Request();
            request.setType(0xa);
            request.setLength(payload.length);
            request.setPayload(payload);

            connection.writeRequest(request);
            BasicReturns basicReturns = waitResponse(basicConsumerArgs.getRid());
            return basicReturns.isOk();

        }
    //  手动确认消息 : 0xb
        public boolean basicAck(String queueName,String messageId) throws IOException {
            BasicAckArgs basicAckArgs = new BasicAckArgs();
            basicAckArgs.setRid(generateRid());
            basicAckArgs.setChannelId(channelId);
            basicAckArgs.setQueueName(queueName);
            basicAckArgs.setMessageId(messageId);
            byte[] payload = BinaryTool.toByte(basicAckArgs);

            Request request = new Request();
            request.setType(0xb);
            request.setLength(payload.length);
            request.setPayload(payload);

            connection.writeRequest(request);
            BasicReturns basicReturns = waitResponse(basicAckArgs.getRid());
            return basicReturns.isOk();
        }

}
